package com.qf.shopsearch.listener;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.qf.common.constant.RabbitMqConstants;
import com.qf.entity.Goods;
import com.rabbitmq.client.Channel;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Component
public class GoodsMQESListener {
    @Value("${elasticsearch.index}")
    private String index;
    @Value("${elasticsearch.type}")
    private String type;
    @Autowired
    private RestHighLevelClient client;
        //之前消费消息都是单线程消费，现在创建线程池多线程消费
        private ExecutorService executorService = Executors.newFixedThreadPool(5);

        //监听的是哪个队列
        @RabbitListener(queues = RabbitMqConstants.GOODS_QUEUE_ES)
        public void addES(Goods goods, Channel channel, Message message) {

        executorService.submit(new Runnable() {
            @Override
            public void run() {
                //goods模块添加sql后，发送消息给mq，监听器消费消息后，同步添加到es中
                ObjectMapper objectMapper = new ObjectMapper();
                try {
                    IndexRequest indexRequest = new IndexRequest(index, type);
                    indexRequest.source(objectMapper.writeValueAsString(goods), XContentType.JSON);
                    client.index(indexRequest, RequestOptions.DEFAULT);
                    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        });
    }

}
